---
title: "Agents"
icon: "robot"
---

## Overview

An **AI Agent** is a system built on language models (LLMs or SLMs) that can solve complex tasks through structured reasoning and autonomous or human-assisted actions. The BeeAI Framework serves as the orchestration layer that enables agents to do this and more:

- Coordinate with LLMs: Manages communication between your agent and language models
- Tool Management: Provides agents with access to external tools and handles their execution
- Response Processing: Processes and validates tool outputs and model responses
- Memory Management: Maintains conversation context and state across interactions
- Error Handling: Manages retries, timeouts, and graceful failure recovery
- Event Orchestration: Emits detailed events for monitoring and debugging agent behavior

Unlike basic chatbots, agents built with this framework can perform multi-step reasoning, use tools to interact with external systems, maintain context across interactions, and adapt based on feedback.
These capabilities make them ideal for planning, research, analysis, and complex execution tasks.

<Tip>
Dive deeper into the concepts behind AI agents in this [research article](https://research.ibm.com/blog/what-are-ai-agents-llm) from IBM.
</Tip>

<Note>
Supported in Python and TypeScript.
</Note>

## Customizing Agent Behavior

You can customize your agent's behavior in several key ways:

### 1. Configuring the Language Model Backend
The backend system manages your connection to different language model providers. The BeeAI Framework supports multiple LLM providers through a unified interface. Learn more about available backends and how to set their parameters in our [backend documentation](./backend).

<CodeGroup>

```py Python [expandable]
from beeai_framework.backend import ChatModel
from beeai_framework.agents.requirement import RequirementAgent

# Using Ollama (local models)
llm = ChatModel.from_name("ollama:granite3.3")

# Using OpenAI
llm = ChatModel.from_name("openai:gpt-5-mini")

# Using Anthropic
llm = ChatModel.from_name("anthropic:claude-sonnet-4")

agent = RequirementAgent(
    llm=llm,
    # ... other configuration
)
```

```ts TypeScript [expandable]
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";
import { ChatModel } from "beeai-framework/backend/chat";

# Using Ollama (local models)
llm = await ChatModel.from_name("ollama:granite3.3")

# Using OpenAI
llm = await ChatModel.from_name("openai:gpt-5-mini")

# Using Anthropic
llm = await ChatModel.from_name("anthropic:claude-sonnet-4")

agent = ToolCallingAgent({
	llm,
	# ... other configuration
})
```

</CodeGroup>

**Backend Features:**
- Unified Interface: Work with different providers using the same API
- Model Parameters: Configure temperature, max tokens, and other model settings
- Provider Support: OpenAI, Anthropic, Ollama, Groq, and more
- Local & Cloud: Support for both local and cloud-hosted models

### 2. Setting the System Prompt

The system prompt defines your agent's behavior, personality, and capabilities. You can configure this through several parameters when initializing an agent:

<CodeGroup>

```py Python [expandable]
agent = RequirementAgent(
    llm=llm,
    role="You are a helpful research assistant specializing in academic papers",
    instructions=[
        "Always provide citations for your sources",
        "Focus on peer-reviewed research when possible",
        "Explain complex concepts in simple terms"
    ],
    notes=[
        "Be especially careful about medical or legal advice",
        "If unsure about a fact, acknowledge the uncertainty"
    ],
    name="Research Assistant",
    description="An AI agent that helps with academic research tasks"
)
```

```ts TypeScript [expandable]
const agent = new ToolCallingAgent({
	llm,
	templates: {
		system: (template) =>
			template.fork((config) => {
				config.defaults.instructions = "You are a helpful assistant that uses tools to answer questions.";
			}),
	}
});
```

</CodeGroup>

**Prompt Parameters:**
- `role`: Defines the agent's persona and primary function
- `instructions`: List of specific behavioral guidelines
- `notes`: Additional context or special considerations
- `name` and `description`: Help identify the agent's purpose and are helpful when using the `HandoffTool`, `Serve` module, and when you need to access agent metadata via `agent.meta`


Setting instructions, notes, role will be integrated into the framework provided system prompt template.  If you want to completely override the framework provided system prompt template, you can provide a custom [prompt template](./templates).

### 3. Configuring Agent Run Options
When executing an agent, you can provide additional options to guide its behavior and execution settings:

#### Setting Execution Settings and Guiding Agent Run Behavior

<CodeGroup>

```py Python [expandable]
response = await agent.run(
    prompt="Analyze the latest AI research trends",
    expected_output="A structured summary with key findings and recommendations",
	# expected_output can also receive a Pydantic Model or a JSON Schema
    backstory="The user is preparing for a conference presentation on AI trends",
    total_max_retries=5,
    max_retries_per_step=2,
    max_iterations=15,
)

#structured outputs can be accessed like this:
print(response.output_structured)
```

```ts TypeScript [expandable]
const schema = z.object({
	firstName: z.string().min(1),
	lastName: z.string().min(1),
	age: z.number().min(1).max(99),
	country: z.string(),
});

const response = await agent.run({
	prompt: "Generate profile of a citizen.",
	expectedOutput: schema,
});
console.info(response.result.text); // JSON text
```

</CodeGroup>

**Available Options:**
- `expected_output`: Guides the agent toward a specific unstructured or structured output format. `output_structured` is defined only when `expected_output` is a Pydantic model or a JSON schema. However, the text representation is always available via `response.output`.
- `backstory`: Provides additional context to help the agent understand the user's situation
- `total_max_retries`: Controls the total number of retry attempts across the entire agent execution
- `max_retries_per_step`: Limits retries for individual steps (like tool calls or model responses)
- `max_iterations`: Sets the maximum number of reasoning cycles the agent can perform

<Tip>
The are defaults set for `max_iterations`, `total_max_retries`, and `max_retries_per_step`, but you can override them by setting your own preferences.
</Tip>


### 4. Adding Tools

Enhance your agent's capabilities by providing it with tools to interact with external systems. Learn more about beeai provided tools and creating custom tools in our [tools documentation](./tools).

<CodeGroup>

```py Python [expandable]
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.weather import OpenMeteoTool

agent = RequirementAgent(
    llm="ollama:granite4",
    tools=[
        DuckDuckGoSearchTool(),
        OpenMeteoTool(),
        # Add more tools as needed
    ]
)
```

```ts TypeScript [expandable]
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { ChatModel } from "beeai-framework/backend/chat";
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";

const llm = await ChatModel.from_name("ollama:granite4");
const agent = new ToolCallingAgent({
	llm,
	tools: [
		new OpenMeteoTool(), // weather tool
	],
});
```

</CodeGroup>

### 5. Configuring Memory

Memory allows your agent to maintain context across multiple interactions. Different memory types serve different use cases. Learn more about our built in options in the [memory documentation](./memory).

<CodeGroup>

```py Python [expandable]
from beeai_framework.memory import TokenMemory, UnconstrainedMemory

# For unlimited conversation history
agent = RequirementAgent(
    llm=llm,
    memory=UnconstrainedMemory()
)
```

```ts TypeScript [expandable]
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { ChatModel } from "beeai-framework/backend/chat";
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";

const llm = await ChatModel.from_name("ollama:granite4");
const agent = new ToolCallingAgent({
	llm,
	memory: new TokenMemory()
});
```

</CodeGroup>

### Additional Agent Options
- [Observability & Debugging](./observability): Monitor agent behavior with detailed event tracking and logging systems
- [MCP (Model Context Protocol)](../integrations/mcp): Connect to external services and data sources
- [A2A (Agent-to-Agent)](../integrations/a2a): Enable multi-agent communication and coordination
- [Caching](./cache): Improve performance by caching LLM responses and tool outputs
- [Event System](./events): Build reactive applications using the comprehensive emitter framework
- [RAG Integration](./rag): Connect your agents to knowledge bases and document stores
- [Serialization](./serialization): Save and restore agent state for persistence and deployment
- [Error Handling](./errors): Implement robust error recovery and debugging strategies


## Agent Types

BeeAI Framework provides several agent implementations:

<Warning>
<strong>Upcoming change:</strong><br />
The <em>Requirement agent</em> will become the primary supported agent. The <em>ReAct</em> and <em>tool-calling</em> agents will not be actively supported.
</Warning>


### Requirement Agent

<Note>
This is the recommended agent. Currently only supported in Python.
</Note>

This agent provides the reliability needed for production scenarios through a rule system that defines execution constraints while keeping problem-solving flexibility intact. Unlike traditional approaches that require complex orchestration code, RequirementAgent uses a declarative interface where you define requirements and let the framework enforce them automatically.

Learn more about RequirementAgent in its dedicated [page](/modules/agents/requirement-agent) or in the [blog post](https://beeai.dev/blog/reliable-ai-agents).

```py Python [expandable]
from beeai_framework.agents.requirement import RequirementAgent
from beeai_framework.agents.requirement.requirements.conditional import ConditionalRequirement
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.think import ThinkTool
from beeai_framework.tools.weather import OpenMeteoTool
from beeai_framework.backend import ChatModel
from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware

agent = RequirementAgent(
    llm=ChatModel.from_name("ollama:granite4:micro"),
    tools=[
        ThinkTool(),             # to reason
        OpenMeteoTool(),         # retrieve weather data
        DuckDuckGoSearchTool()   # search web
    ],
    instructions="Plan activities for a given destination based on current weather and events.",
    requirements=[
        # Force thinking first
        ConditionalRequirement(ThinkTool, force_at_step=1),
        # Search only after getting weather and at least once
        ConditionalRequirement(DuckDuckGoSearchTool, only_after=[OpenMeteoTool], min_invocations=1),
        # Weather tool be used at least once but not consecutively
        ConditionalRequirement(OpenMeteoTool, consecutive_allowed=False, min_invocations=1),
    ]
)

# Run with execution logging
response = await agent.run("What to do in Boston?").middleware(GlobalTrajectoryMiddleware())
print(f"Final Answer: {response.answer.text}")

```


### ReAct Agent

<Tip>
The ReAct Agent is available in both Python and TypeScript, but no longer actively supported.
</Tip>


The ReActAgent implements the ReAct ([Reasoning and Acting](https://arxiv.org/abs/2210.03629)) pattern, which structures agent behavior into a cyclical process of reasoning, action, and observation.

This pattern allows agents to reason about a task, take actions using tools, observe results, and continue reasoning until reaching a conclusion.

Let's see how a ReActAgent approaches a simple question:

**Input prompt:** "What is the current weather in Las Vegas?"

**First iteration:**

```log
thought: I need to retrieve the current weather in Las Vegas. I can use the OpenMeteo function to get the current weather forecast for a location.
tool_name: OpenMeteo
tool_input: {"location": {"name": "Las Vegas"}, "start_date": "2024-10-17", "end_date": "2024-10-17", "temperature_unit": "celsius"}
```

**Second iteration:**

```log
thought: I have the current weather in Las Vegas in Celsius.
final_answer: The current weather in Las Vegas is 20.5°C with an apparent temperature of 18.3°C.
```

<Note>
During execution, the agent emits partial updates as it generates each line, followed by complete updates. Updates follow a strict order: first all partial updates for "thought," then a complete "thought" update, then moving to the next component.
</Note>

<CodeGroup>
{/* <!-- embedme python/examples/agents/react.py --> */}
```py Python [expandable]
import asyncio
import logging
import os
import sys
import tempfile
import traceback
from typing import Any

from dotenv import load_dotenv

from beeai_framework.agents.react import ReActAgent
from beeai_framework.backend import ChatModel, ChatModelParameters
from beeai_framework.emitter import EmitterOptions, EventMeta
from beeai_framework.errors import FrameworkError
from beeai_framework.logger import Logger
from beeai_framework.memory import TokenMemory
from beeai_framework.tools import AnyTool
from beeai_framework.tools.code import LocalPythonStorage, PythonTool
from beeai_framework.tools.search.duckduckgo import DuckDuckGoSearchTool
from beeai_framework.tools.search.wikipedia import WikipediaTool
from beeai_framework.tools.weather import OpenMeteoTool
from examples.helpers.io import ConsoleReader

# Load environment variables
load_dotenv()

# Configure logging - using DEBUG instead of trace
logger = Logger("app", level=logging.DEBUG)

reader = ConsoleReader()


def create_agent() -> ReActAgent:
    """Create and configure the agent with tools and LLM"""

    # Other models to try:
    # "llama3.1"
    # "granite3.3"
    # "deepseek-r1"
    # ensure the model is pulled before running
    llm = ChatModel.from_name(
        "ollama:granite4:micro",
        ChatModelParameters(temperature=0),
    )

    # Configure tools
    tools: list[AnyTool] = [
        WikipediaTool(),
        OpenMeteoTool(),
        DuckDuckGoSearchTool(),
    ]

    # Add code interpreter tool if URL is configured
    code_interpreter_url = os.getenv("CODE_INTERPRETER_URL")
    if code_interpreter_url:
        tools.append(
            PythonTool(
                code_interpreter_url,
                LocalPythonStorage(
                    local_working_dir=tempfile.mkdtemp("code_interpreter_source"),
                    interpreter_working_dir=os.getenv("CODE_INTERPRETER_TMPDIR", "./tmp/code_interpreter_target"),
                ),
            )
        )

    # Create agent with memory and tools
    agent = ReActAgent(llm=llm, tools=tools, memory=TokenMemory(llm))

    return agent


def process_agent_events(data: Any, event: EventMeta) -> None:
    """Process agent events and log appropriately"""

    if event.name == "error":
        reader.write("Agent 🤖 : ", FrameworkError.ensure(data.error).explain())
    elif event.name == "retry":
        reader.write("Agent 🤖 : ", "retrying the action...")
    elif event.name == "update":
        reader.write(f"Agent({data.update.key}) 🤖 : ", data.update.parsed_value)
    elif event.name == "start":
        reader.write("Agent 🤖 : ", "starting new iteration")
    elif event.name == "success":
        reader.write("Agent 🤖 : ", "success")


async def main() -> None:
    """Main application loop"""

    # Create agent
    agent = create_agent()

    # Log code interpreter status if configured
    code_interpreter_url = os.getenv("CODE_INTERPRETER_URL")
    if code_interpreter_url:
        reader.write(
            "🛠️ System: ",
            f"The code interpreter tool is enabled. Please ensure that it is running on {code_interpreter_url}",
        )

    reader.write("🛠️ System: ", "Agent initialized with Wikipedia, DuckDuckGo, and Weather tools.")

    # Main interaction loop with user input
    for prompt in reader:
        # Run agent with the prompt
        response = await agent.run(
            prompt,
            max_retries_per_step=3,
            total_max_retries=10,
            max_iterations=20,
        ).on("*", process_agent_events, EmitterOptions(match_nested=False))

        reader.write("Agent 🤖 : ", response.last_message.text)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

```
{/* <!-- embedme typescript/examples/agents/react.ts --> */}
```ts TypeScript [expandable]
import "dotenv/config.js";
import { ReActAgent } from "beeai-framework/agents/react/agent";
import { createConsoleReader } from "../helpers/io.js";
import { FrameworkError } from "beeai-framework/errors";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { Logger } from "beeai-framework/logger/logger";
import { PythonTool } from "beeai-framework/tools/python/python";
import { LocalPythonStorage } from "beeai-framework/tools/python/storage";
import { DuckDuckGoSearchTool } from "beeai-framework/tools/search/duckDuckGoSearch";
import { WikipediaTool } from "beeai-framework/tools/search/wikipedia";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { dirname } from "node:path";
import { fileURLToPath } from "node:url";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";

Logger.root.level = "silent"; // disable internal logs
const logger = new Logger({ name: "app", level: "trace" });

// Other models to try:
// "llama3.1:70b"
// "granite3.3"
// "deepseek-r1:32b"
// ensure the model is pulled before running
const llm = new OllamaChatModel("llama3.1");

const codeInterpreterUrl = process.env.CODE_INTERPRETER_URL;
const __dirname = dirname(fileURLToPath(import.meta.url));

const codeInterpreterTmpdir =
  process.env.CODE_INTERPRETER_TMPDIR ?? "./examples/tmp/code_interpreter";
const localTmpdir = process.env.LOCAL_TMPDIR ?? "./examples/tmp/local";

const agent = new ReActAgent({
  llm,
  memory: new TokenMemory(),
  tools: [
    new DuckDuckGoSearchTool(),
    // new WebCrawlerTool(), // HTML web page crawler
    new WikipediaTool(),
    new OpenMeteoTool(), // weather tool
    // new ArXivTool(), // research papers
    // new DynamicTool() // custom python tool
    ...(codeInterpreterUrl
      ? [
          new PythonTool({
            codeInterpreter: { url: codeInterpreterUrl },
            storage: new LocalPythonStorage({
              interpreterWorkingDir: `${__dirname}/../../${codeInterpreterTmpdir}`,
              localWorkingDir: `${__dirname}/../../${localTmpdir}`,
            }),
          }),
        ]
      : []),
  ],
});

const reader = createConsoleReader();
if (codeInterpreterUrl) {
  reader.write(
    "🛠️ System",
    `The code interpreter tool is enabled. Please ensure that it is running on ${codeInterpreterUrl}`,
  );
}

try {
  for await (const { prompt } of reader) {
    const response = await agent
      .run(
        { prompt },
        {
          execution: {
            maxRetriesPerStep: 3,
            totalMaxRetries: 10,
            maxIterations: 20,
          },
        },
      )
      .observe((emitter) => {
        // emitter.on("start", () => {
        //   reader.write(`Agent 🤖 : `, "starting new iteration");
        // });
        emitter.on("error", ({ error }) => {
          reader.write(`Agent 🤖 : `, FrameworkError.ensure(error).dump());
        });
        emitter.on("retry", () => {
          reader.write(`Agent 🤖 : `, "retrying the action...");
        });
        emitter.on("update", async ({ data, update, meta }) => {
          // log 'data' to see the whole state
          // to log only valid runs (no errors), check if meta.success === true
          reader.write(`Agent (${update.key}) 🤖 : `, update.value);
        });
        emitter.on("partialUpdate", ({ data, update, meta }) => {
          // ideal for streaming (line by line)
          // log 'data' to see the whole state
          // to log only valid runs (no errors), check if meta.success === true
          // reader.write(`Agent (partial ${update.key}) 🤖 : `, update.value);
        });

        // To observe all events (uncomment following block)
        // emitter.match("*.*", async (data: unknown, event) => {
        //   logger.trace(event, `Received event "${event.path}"`);
        // });

        // To get raw LLM input (uncomment following block)
        // emitter.match(
        //   (event) => event.creator === llm && event.name === "start",
        //   async (data: InferCallbackValue<GenerateEvents["start"]>, event) => {
        //     logger.trace(
        //       event,
        //       [
        //         `Received LLM event "${event.path}"`,
        //         JSON.stringify(data.input), // array of messages
        //       ].join("\n"),
        //     );
        //   },
        // );
      });

    reader.write(`Agent 🤖 : `, response.result.text);
  }
} catch (error) {
  logger.error(FrameworkError.ensure(error).dump());
} finally {
  reader.close();
}

```

</CodeGroup>

### Tool Calling Agent

<Tip>
The Tool Calling Agent is available in both Python and TypeScript, but no longer actively supported.
</Tip>

The ToolCallingAgent is optimized for scenarios where tool usage is the primary focus. It handles tool calls more efficiently and can execute multiple tools in parallel.

<CodeGroup>
{/* <!-- embedme python/examples/agents/tool_calling.py --> */}
```py Python [expandable]
import asyncio
import logging
import sys
import traceback
from typing import Any

from dotenv import load_dotenv

from beeai_framework.agents.tool_calling import ToolCallingAgent
from beeai_framework.backend import ChatModel
from beeai_framework.emitter import EventMeta
from beeai_framework.errors import FrameworkError
from beeai_framework.logger import Logger
from beeai_framework.memory import UnconstrainedMemory
from beeai_framework.tools.weather import OpenMeteoTool
from examples.helpers.io import ConsoleReader

# Load environment variables
load_dotenv()

# Configure logging - using DEBUG instead of trace
logger = Logger("app", level=logging.DEBUG)

reader = ConsoleReader()


def process_agent_events(data: Any, event: EventMeta) -> None:
    """Process agent events and log appropriately"""

    if event.name == "start":
        reader.write("Agent (debug) 🤖 : ", "starting new iteration")
    elif event.name == "success":
        reader.write("Agent (debug) 🤖 : ", data.state.memory.messages[-1])


async def main() -> None:
    """Main application loop"""

    # Create agent
    agent = ToolCallingAgent(
        llm=ChatModel.from_name("ollama:llama3.1"), memory=UnconstrainedMemory(), tools=[OpenMeteoTool()]
    )

    # Main interaction loop with user input
    for prompt in reader:
        response = await agent.run(prompt).on("*", process_agent_events)
        reader.write("Agent 🤖 : ", response.last_message.text)

    print("======DONE (showing the full message history)=======")

    messages = response.state.memory.messages
    for msg in messages:
        print(msg)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

```
{/* <!-- embedme typescript/examples/agents/toolCalling/agent.ts --> */}
```ts TypeScript [expandable]
import "dotenv/config.js";
import { createConsoleReader } from "../../helpers/io.js";
import { FrameworkError } from "beeai-framework/errors";
import { TokenMemory } from "beeai-framework/memory/tokenMemory";
import { Logger } from "beeai-framework/logger/logger";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";
import { ToolCallingAgent } from "beeai-framework/agents/toolCalling/agent";

Logger.root.level = "silent"; // disable internal logs
const logger = new Logger({ name: "app", level: "trace" });

// Other models to try:
// "llama3.1:70b"
// "granite3.3"
// "deepseek-r1:32b"
// ensure the model is pulled before running
const llm = new OllamaChatModel("llama3.1");

const agent = new ToolCallingAgent({
  llm,
  memory: new TokenMemory(),
  templates: {
    system: (template) =>
      template.fork((config) => {
        config.defaults.instructions =
          "You are a helpful assistant that uses tools to answer questions.";
      }),
  },
  tools: [
    new OpenMeteoTool(), // weather tool
  ],
});

const reader = createConsoleReader();

try {
  for await (const { prompt } of reader) {
    let messagesCount = agent.memory.messages.length + 1;

    const response = await agent.run({ prompt }).observe((emitter) => {
      emitter.on("success", async ({ state }) => {
        const newMessages = state.memory.messages.slice(messagesCount);
        messagesCount += newMessages.length;

        reader.write(
          `Agent (${newMessages.length} new messages) 🤖 :\n`,
          newMessages.map((msg) => `-> ${JSON.stringify(msg.toPlain())}`).join("\n"),
        );
      });

      // To observe all events (uncomment following block)
      // emitter.match("*.*", async (data: unknown, event) => {
      //   logger.trace(event, `Received event "${event.path}"`);
      // }, {
      //   matchNested: true
      // });

      // To get raw LLM input (uncomment following block)
      // emitter.match(
      //   (event) => event.creator === llm && event.name === "start",
      //   async (data: InferCallbackValue<GenerateEvents["start"]>, event) => {
      //     logger.trace(
      //       event,
      //       [
      //         `Received LLM event "${event.path}"`,
      //         JSON.stringify(data.input), // array of messages
      //       ].join("\n"),
      //     );
      //   },
      // );
    });

    reader.write(`Agent 🤖 : `, response.result.text);
  }
} catch (error) {
  logger.error(FrameworkError.ensure(error).dump());
} finally {
  reader.close();
}

```

</CodeGroup>

### Custom Agent

For advanced use cases, you can create your own agent implementation by extending the `BaseAgent` class.

<CodeGroup>

```py Python [expandable]
import asyncio
import sys
import traceback
from typing import Unpack

from pydantic import BaseModel, Field

from beeai_framework.adapters.ollama import OllamaChatModel
from beeai_framework.agents import AgentMeta, AgentOptions, AgentOutput, BaseAgent
from beeai_framework.backend import AnyMessage, AssistantMessage, ChatModel, SystemMessage, UserMessage
from beeai_framework.context import RunContext
from beeai_framework.emitter import Emitter
from beeai_framework.errors import FrameworkError
from beeai_framework.memory import BaseMemory, UnconstrainedMemory
from beeai_framework.runnable import runnable_entry


class State(BaseModel):
    thought: str
    final_answer: str


class CustomAgent(BaseAgent):
    def __init__(self, llm: ChatModel, memory: BaseMemory) -> None:
        super().__init__()
        self.model = llm
        self._memory = memory

    @property
    def memory(self) -> BaseMemory:
        return self._memory

    @memory.setter
    def memory(self, memory: BaseMemory) -> None:
        self._memory = memory

    def _create_emitter(self) -> Emitter:
        return Emitter.root().child(
            namespace=["agent", "custom"],
            creator=self,
        )

    @runnable_entry
    async def run(self, input: str | list[AnyMessage], /, **kwargs: Unpack[AgentOptions]) -> AgentOutput:
        async def handler(context: RunContext) -> AgentOutput:
            class CustomSchema(BaseModel):
                thought: str = Field(description="Describe your thought process before coming with a final answer")
                final_answer: str = Field(
                    description="Here you should provide concise answer to the original question."
                )

            response = await self.model.run(
                [
                    SystemMessage("You are a helpful assistant. Always use JSON format for your responses."),
                    *(self.memory.messages if self.memory is not None else []),
                    *([UserMessage(input)] if isinstance(input, str) else input),
                ],
                response_format=CustomSchema,
                max_retries=kwargs.get("total_max_retries", 3),
                signal=context.signal,
            )
            assert isinstance(response.output_structured, CustomSchema)

            result = AssistantMessage(response.output_structured.final_answer)
            await self.memory.add(result) if self.memory else None

            return AgentOutput(
                output=[result],
                context={
                    "state": State(
                        thought=response.output_structured.thought,
                        final_answer=response.output_structured.final_answer,
                    )
                },
            )

        return await handler(RunContext.get())

    @property
    def meta(self) -> AgentMeta:
        return AgentMeta(
            name="CustomAgent",
            description="Custom Agent is a simple LLM agent.",
            tools=[],
        )


async def main() -> None:
    agent = CustomAgent(
        llm=OllamaChatModel("granite3.3"),
        memory=UnconstrainedMemory(),
    )

    response = await agent.run([UserMessage("Why is the sky blue?")])
    print(response.context.get("state"))


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

```

```ts TypeScript [expandable]
import { BaseAgent, BaseAgentRunOptions } from "beeai-framework/agents/base";
import {
  AssistantMessage,
  Message,
  SystemMessage,
  UserMessage,
} from "beeai-framework/backend/message";
import { Emitter } from "beeai-framework/emitter/emitter";
import { GetRunContext } from "beeai-framework/context";
import { z } from "zod";
import { AgentMeta } from "beeai-framework/agents/types";
import { BaseMemory } from "beeai-framework/memory/base";
import { UnconstrainedMemory } from "beeai-framework/memory/unconstrainedMemory";
import { ChatModel } from "beeai-framework/backend/chat";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";

interface RunInput {
  message: Message;
}

interface RunOutput {
  message: Message;
  state: {
    thought: string;
    final_answer: string;
  };
}

interface RunOptions extends BaseAgentRunOptions {
  maxRetries?: number;
}

interface AgentInput {
  llm: ChatModel;
  memory: BaseMemory;
}

export class CustomAgent extends BaseAgent<RunInput, RunOutput, RunOptions> {
  public readonly memory: BaseMemory;
  protected readonly model: ChatModel;
  public emitter = Emitter.root.child({
    namespace: ["agent", "custom"],
    creator: this,
  });

  constructor(input: AgentInput) {
    super();
    this.model = input.llm;
    this.memory = input.memory;
  }

  protected async _run(
    input: RunInput,
    options: RunOptions,
    run: GetRunContext<this>,
  ): Promise<RunOutput> {
    const response = await this.model.createStructure({
      schema: z.object({
        thought: z
          .string()
          .describe("Describe your thought process before coming with a final answer"),
        final_answer: z
          .string()
          .describe("Here you should provide concise answer to the original question."),
      }),
      messages: [
        new SystemMessage("You are a helpful assistant. Always use JSON format for you responses."),
        ...this.memory.messages,
        input.message,
      ],
      maxRetries: options?.maxRetries,
      abortSignal: run.signal,
    });

    const result = new AssistantMessage(response.object.final_answer);
    await this.memory.add(result);

    return {
      message: result,
      state: response.object,
    };
  }

  public get meta(): AgentMeta {
    return {
      name: "CustomAgent",
      description: "Custom Agent is a simple LLM agent.",
      tools: [],
    };
  }

  createSnapshot() {
    return {
      ...super.createSnapshot(),
      emitter: this.emitter,
      memory: this.memory,
    };
  }

  loadSnapshot(snapshot: ReturnType<typeof this.createSnapshot>) {
    Object.assign(this, snapshot);
  }
}

const agent = new CustomAgent({
  llm: new OllamaChatModel("granite3.3"),
  memory: new UnconstrainedMemory(),
});

const response = await agent.run({
  message: new UserMessage("Why is the sky blue?"),
});
console.info(response.state);

```

</CodeGroup>

## Multi-Agent Hand-offs

Create a team of specialized agents that can collaborate:

<CodeGroup>

	{/* <!-- embedme python/examples/agents/requirement/handoff.py --> */}
	```py Python [expandable]
	import asyncio
	
	from beeai_framework.agents.requirement import RequirementAgent
	from beeai_framework.agents.requirement.requirements.conditional import ConditionalRequirement
	from beeai_framework.backend import ChatModel
	from beeai_framework.errors import FrameworkError
	from beeai_framework.middleware.trajectory import GlobalTrajectoryMiddleware
	from beeai_framework.tools import Tool
	from beeai_framework.tools.handoff import HandoffTool
	from beeai_framework.tools.search.wikipedia import WikipediaTool
	from beeai_framework.tools.think import ThinkTool
	from beeai_framework.tools.weather import OpenMeteoTool
	
	
	async def main() -> None:
	    knowledge_agent = RequirementAgent(
	        llm=ChatModel.from_name("ollama:granite4:micro"),
	        tools=[ThinkTool(), WikipediaTool()],
	        requirements=[ConditionalRequirement(ThinkTool, force_at_step=1)],
	        role="Knowledge Specialist",
	        instructions="Provide answers to general questions about the world.",
	    )
	
	    weather_agent = RequirementAgent(
	        llm=ChatModel.from_name("ollama:granite4:micro"),
	        tools=[OpenMeteoTool()],
	        role="Weather Specialist",
	        instructions="Provide weather forecast for a given destination.",
	    )
	
	    main_agent = RequirementAgent(
	        name="MainAgent",
	        llm=ChatModel.from_name("ollama:granite4:micro"),
	        tools=[
	            ThinkTool(),
	            HandoffTool(
	                knowledge_agent,
	                name="KnowledgeLookup",
	                description="Consult the Knowledge Agent for general questions.",
	            ),
	            HandoffTool(
	                weather_agent,
	                name="WeatherLookup",
	                description="Consult the Weather Agent for forecasts.",
	            ),
	        ],
	        requirements=[ConditionalRequirement(ThinkTool, force_at_step=1)],
	        # Log all tool calls to the console for easier debugging
	        middlewares=[GlobalTrajectoryMiddleware(included=[Tool])],
	    )
	
	    question = "If I travel to Rome next weekend, what should I expect in terms of weather, and also tell me one famous historical landmark there?"
	    print(f"User: {question}")
	
	    try:
	        response = await main_agent.run(question, expected_output="Helpful and clear response.")
	        print("Agent:", response.last_message.text)
	    except FrameworkError as err:
	        print("Error:", err.explain())
	
	
	if __name__ == "__main__":
	    asyncio.run(main())
	
	```

	```ts TypeScript [expandable]
	COMING SOON
	```

</CodeGroup>

## Workflows

<Warning>
<strong>Upcoming change:</strong><br />
Workflows are under construction to support more dynamic multi-agent patterns. If you'd like to participate in shaping the vision, contribute to the discussion in this [V2 Workflow Proposal](https://github.com/i-am-bee/beeai-framework/discussions/1005).
</Warning>

For complex applications, you can create multi-agent workflows where specialized agents collaborate.

<CodeGroup>

```py Python [expandable]
import asyncio
import sys
import traceback

from beeai_framework.backend import ChatModel
from beeai_framework.emitter import EmitterOptions
from beeai_framework.errors import FrameworkError
from beeai_framework.tools.search.wikipedia import WikipediaTool
from beeai_framework.tools.weather import OpenMeteoTool
from beeai_framework.workflows.agent import AgentWorkflow, AgentWorkflowInput
from examples.helpers.io import ConsoleReader


async def main() -> None:
    llm = ChatModel.from_name("ollama:llama3.1")
    workflow = AgentWorkflow(name="Smart assistant")

    workflow.add_agent(
        name="Researcher",
        role="A diligent researcher.",
        instructions="You look up and provide information about a specific topic.",
        tools=[WikipediaTool()],
        llm=llm,
    )

    workflow.add_agent(
        name="WeatherForecaster",
        role="A weather reporter.",
        instructions="You provide detailed weather reports.",
        tools=[OpenMeteoTool()],
        llm=llm,
    )

    workflow.add_agent(
        name="DataSynthesizer",
        role="A meticulous and creative data synthesizer",
        instructions="You can combine disparate information into a final coherent summary.",
        llm=llm,
    )

    reader = ConsoleReader()

    reader.write("Assistant 🤖 : ", "What location do you want to learn about?")
    for prompt in reader:
        await (
            workflow.run(
                inputs=[
                    AgentWorkflowInput(prompt="Provide a short history of the location.", context=prompt),
                    AgentWorkflowInput(
                        prompt="Provide a comprehensive weather summary for the location today.",
                        expected_output="Essential weather details such as chance of rain, temperature and wind. Only report information that is available.",
                    ),
                    AgentWorkflowInput(
                        prompt="Summarize the historical and weather data for the location.",
                        expected_output="A paragraph that describes the history of the location, followed by the current weather conditions.",
                    ),
                ]
            )
            .on(
                # Event Matcher -> match agent's 'success' events
                lambda event: isinstance(event.creator, ChatModel) and event.name == "success",
                # log data to the console
                lambda data, event: reader.write(
                    "->Got response from the LLM",
                    "  \n->".join([str(message.content[0].model_dump()) for message in data.value.messages]),
                ),
                EmitterOptions(match_nested=True),
            )
            .on(
                "success",
                lambda data, event: reader.write(
                    f"->Step '{data.step}' has been completed with the following outcome."
                    f"\n\n{data.state.final_answer}\n\n",
                    data.model_dump(exclude={"data"}),
                ),
            )
        )
        reader.write("Assistant 🤖 : ", "What location do you want to learn about?")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except FrameworkError as e:
        traceback.print_exc()
        sys.exit(e.explain())

```

```ts TypeScript [expandable]
import "dotenv/config";
import { createConsoleReader } from "examples/helpers/io.js";
import { OpenMeteoTool } from "beeai-framework/tools/weather/openMeteo";
import { WikipediaTool } from "beeai-framework/tools/search/wikipedia";
import { AgentWorkflow } from "beeai-framework/workflows/agent";
import { OllamaChatModel } from "beeai-framework/adapters/ollama/backend/chat";

const workflow = new AgentWorkflow("Smart assistant");
const llm = new OllamaChatModel("llama3.1");

workflow.addAgent({
  name: "Researcher",
  role: "A diligent researcher",
  instructions: "You look up and provide information about a specific topic.",
  tools: [new WikipediaTool()],
  llm,
});
workflow.addAgent({
  name: "WeatherForecaster",
  role: "A weather reporter",
  instructions: "You provide detailed weather reports.",
  tools: [new OpenMeteoTool()],
  llm,
});
workflow.addAgent({
  name: "DataSynthesizer",
  role: "A meticulous and creative data synthesizer",
  instructions: "You can combine disparate information into a final coherent summary.",
  llm,
});

const reader = createConsoleReader();
reader.write("Assistant 🤖 : ", "What location do you want to learn about?");
for await (const { prompt } of reader) {
  const { result } = await workflow
    .run([
      { prompt: "Provide a short history of the location.", context: prompt },
      {
        prompt: "Provide a comprehensive weather summary for the location today.",
        expectedOutput:
          "Essential weather details such as chance of rain, temperature and wind. Only report information that is available.",
      },
      {
        prompt: "Summarize the historical and weather data for the location.",
        expectedOutput:
          "A paragraph that describes the history of the location, followed by the current weather conditions.",
      },
    ])
    .observe((emitter) => {
      emitter.on("success", (data) => {
        reader.write(
          `Step '${data.step}' has been completed with the following outcome:\n`,
          data.state?.finalAnswer ?? "-",
        );
      });
    });

  reader.write(`Assistant 🤖`, result.finalAnswer);
  reader.write("Assistant 🤖 : ", "What location do you want to learn about?");
}

```

</CodeGroup>

## Examples

<CardGroup cols={2}>
  <Card title="Python" icon="python" href="https://github.com/i-am-bee/beeai-framework/tree/main/python/examples/agents">
    Explore reference agent implementations in Python
  </Card>
  <Card title="TypeScript" icon="js" href="https://github.com/i-am-bee/beeai-framework/tree/main/typescript/examples/agents">
    Explore reference agent implementations in TypeScript
  </Card>
</CardGroup>
